package com.junmo.www.RxJava;


import io.reactivex.*;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.observables.ConnectableObservable;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.subjects.AsyncSubject;
import io.reactivex.subjects.BehaviorSubject;
import io.reactivex.subjects.PublishSubject;
import io.reactivex.subjects.ReplaySubject;
import org.testng.annotations.Test;

import java.util.concurrent.TimeUnit;

public class MainTest {
    @Test
    public  void test() {

    }
    @Test
    public  void test19() {
        ReplaySubject<String> subject = ReplaySubject.createWithSize(2) ;
        subject.onNext ("replaySubject1 ");
        subject.onNext ("replaySubject2 ");
        subject.subscribe(
                new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System. out . println (" replaySubject :"+s) ;
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        System.out .println (" replaySubject onError ");
                    }
                }, new Action() {
                    @Override
                    public void run() throws Exception {
                        System.out .println (" replaySubject: complete ");
                    }
                });
        subject.onNext ("replaySubject3 ");
        subject.onNext ("replaySubject4");
    }

    @Test
    public  void test18() {
        ReplaySubject<String> subject = ReplaySubject. create ( ) ;
        subject.onNext ("replaySubject1 ");
        subject.onNext ("replaySubject2 ");
        subject.subscribe(
                new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System. out . println (" replaySubject :"+s) ;
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        System.out .println (" replaySubject onError ");
                    }
                }, new Action() {
                    @Override
                    public void run() throws Exception {
                        System.out .println (" replaySubject: complete ");
                    }
                });
        subject.onNext ("replaySubject3 ");
        subject.onNext ("replaySubject4");
    }

    @Test
    public  void test17() {
        BehaviorSubject<String> subject = BehaviorSubject . createDefault (" behaviorSubjectl" ) ;
        subject.onNext ("behaviorSubject2 ");
        subject.subscribe(
                new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System. out . println (" behaviorSubject :"+s) ;
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        System.out .println (" behaviorSubject onError ");
                    }
                }, new Action() {
                    @Override
                    public void run() throws Exception {
                        System.out .println (" behaviorSubject: complete ");
                    }
                });
        subject.onNext ("behaviorSubject3 ");
        subject.onNext ("behaviorSubject4");
    }

    @Test
    public  void test16() {
        BehaviorSubject<String> subject = BehaviorSubject . createDefault (" behaviorSubjectl" ) ;
        subject.subscribe(
                new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System. out . println (" behaviorSubject :"+s) ;
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        System.out .println (" behaviorSubject onError ");
                    }
                }, new Action() {
                    @Override
                    public void run() throws Exception {
                        System.out .println (" behaviorSubject: complete ");
                    }
                });
        subject.onNext ("behaviorSubject2 ");
        subject.onNext ("behaviorSubject3");
    }


    @Test// 不会有任何输出
    public  void test15() {
        AsyncSubject<String> subject= AsyncSubject.create();
        subject . onNext (" asyncSubject1");
        subject . onNext (" asyncSubject2 ") ;

        subject.subscribe(
                new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System. out . println (" asyncSubject :"+s) ;
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        System.out .println (" asyncSubject onError ");
                    }
                }, new Action() {
                    @Override
                    public void run() throws Exception {
                        System.out .println (" asyncSub ject: complete ");
                    }
                });
        subject.onNext (" a syncSubject3 ");
        subject . onNext (" a syncSubject4 ");
    }

    @Test
    public  void test14() {
        AsyncSubject<String> subject= AsyncSubject.create();
        subject . onNext (" asyncSubject1");
        subject . onNext (" asyncSubject2 ") ;

        subject.subscribe(
                new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System. out . println (" asyncSubject :"+s) ;
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        System.out .println (" asyncSubject onError ");
                    }
                }, new Action() {
                    @Override
                    public void run() throws Exception {
                        System.out .println (" asyncSub ject: complete ");
                    }
                });
        subject.onNext (" a syncSubject3 ");
        subject . onNext (" a syncSubject4 ");
        subject . onComplete() ;
    }

    @Test
    public  void test13() {
        AsyncSubject<String> subject= AsyncSubject.create();
        subject . onNext (" asyncSubject1");
        subject . onNext (" asyncSubject2 ") ;
        subject . onComplete() ;
        subject.subscribe(
           new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                System. out . println (" asyncSubject :"+s) ;
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                System.out .println (" asyncSubject onError ");
            }
        }, new Action() {
            @Override
            public void run() throws Exception {
                System.out .println (" asyncSub ject: complete ");
            }
        });
        subject.onNext (" a syncSubject3 ");
        subject . onNext (" a syncSubject4 ");
    }


    @Test
    public  void test12() {
       Maybe.create(new MaybeOnSubscribe<String>() {
           @Override
           public void subscribe(@NonNull MaybeEmitter<String> e) throws Exception {
//               e.onComplete() ; // 若先调用了此方法，就不会发送任何数据
               e. onSuccess ("testA" ) ;
               e. onSuccess ("testB" ) ; //不会被接收，因为只能发射 0 或 1 个数据，
           }
       }).subscribe(new Consumer<String>() {
           @Override
           public void accept(String s) throws Exception {
               System.out . println ("s = " +s) ;
           }
       });
    }
    @Test
    public  void test11() {
        Completable.fromAction(new Action() {
            @Override
            public void run() throws Exception {

            }
        });
    }
    @Test
    public  void test10() {
        Completable.create(new CompletableOnSubscribe() {
            @Override
            public void subscribe(@NonNull CompletableEmitter completableEmitter) throws Exception {

            }
        });
    }

    @Test
    public  void test09() {
        Single.create(new SingleOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull SingleEmitter<String> singleEmitter) throws Exception {

            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {

            }
        });
    }

    @Test
    public  void test08() throws InterruptedException {
        Consumer<Long> subscriber1 = aLong -> System.out.println("subscriber1 ：" +aLong);
        Consumer<Long> subscriber2 = aLong -> System.out.println("    subscriber2 ：" +aLong);


        ConnectableObservable<Long> connectableObservable= Observable.create(new ObservableOnSubscribe<Long>(){
            @Override
            public void subscribe(@NonNull ObservableEmitter<Long> e) throws Exception {
                Observable.interval(10 , TimeUnit.MILLISECONDS, Schedulers.computation())
                        . take(Integer.MAX_VALUE). subscribe (e ::onNext);
            }
        }).observeOn (Schedulers.newThread()).publish();
        connectableObservable.connect();

        Observable<Long> observable= connectableObservable.refCount ();

        Disposable disposable1= observable .subscribe(subscriber1) ;
        Disposable disposable2= observable.subscribe(subscriber2) ;

//        Thread.sleep(100L) ;
        Thread.sleep(50L) ;

        disposable1.dispose ();
        disposable2.dispose();
        System .out .println ("=================重新开始数据流");

        disposable1= observable .subscribe(subscriber1) ;
        disposable2= observable.subscribe(subscriber2) ;

        Thread.sleep(50L) ;
    }

    @Test
    public  void test07() throws InterruptedException {
        Consumer<Long> subscriber1 = aLong -> System.out.println("subscriber1 ：" +aLong);
        Consumer<Long> subscriber2 = aLong -> System.out.println("    subscriber2 ：" +aLong);
        Consumer<Long> subscriber3 = aLong -> System.out.println("        subscriber3 ：" +aLong);

        Observable<Long> observable= Observable.create(new ObservableOnSubscribe<Long>(){
            @Override
            public void subscribe(@NonNull ObservableEmitter<Long> e) throws Exception {
                Observable.interval(10 , TimeUnit.MILLISECONDS, Schedulers.computation())
                        . take(Integer.MAX_VALUE). subscribe (e ::onNext);
            }
        }).observeOn (Schedulers.newThread());
        PublishSubject<Long> subject = PublishSubject.create() ;
        observable.subscribe( subject) ;

        observable .subscribe(subscriber1) ;
        observable.subscribe(subscriber2) ;

        Thread.sleep(20L) ;

        observable.subscribe(subscriber3) ;

        Thread.sleep(100L) ;
    }


    @Test
    public  void test06() throws InterruptedException {
        Consumer<Long> subscriber1 =new Consumer<Long>(){
            @Override
            public void accept(Long aLong) throws Exception {
                System.out.println("subscriber1 ：" +aLong);
            }
        };
        Consumer<Long> subscriber2 =new Consumer<Long>(){
            @Override
            public void accept(Long aLong) throws Exception {
                System.out.println("    subscriber2 ：" +aLong);
            }
        };
        Consumer<Long> subscriber3 =new Consumer<Long>(){
            @Override
            public void accept(Long aLong) throws Exception {
                System.out.println("        subscriber3 ：" +aLong);
            }
        };

        ConnectableObservable<Long> observable= Observable.create(new ObservableOnSubscribe<Long>(){
            @Override
            public void subscribe(@NonNull ObservableEmitter<Long> e) throws Exception {
                Observable.interval(10 , TimeUnit.MILLISECONDS, Schedulers.computation())
                        . take(Integer.MAX_VALUE). subscribe (e ::onNext);
            }
        }).observeOn (Schedulers.newThread())
          .publish();
        observable.connect();

        observable .subscribe(subscriber1) ;
        observable.subscribe(subscriber2) ;

//        Thread.sleep(100L) ;
        Thread.sleep(50L) ;

        observable.subscribe(subscriber3) ;

        Thread.sleep(100L) ;
    }


    @Test
    public  void test05() throws InterruptedException {
        Consumer<Long> subscriber1 =new Consumer<Long>(){

            @Override
            public void accept(Long aLong) throws Exception {
                System.out.println("subscriber1 ：" +aLong);
            }
        };
        Consumer<Long> subscriber2 =new Consumer<Long>(){

            @Override
            public void accept(Long aLong) throws Exception {
                System.out.println("    subscriber2 ：" +aLong);
            }
        };
        Observable<Long> observable= Observable.create(new ObservableOnSubscribe<Long>(){
            @Override
            public void subscribe(@NonNull ObservableEmitter<Long> e) throws Exception {
                Observable.interval(10 , TimeUnit.MILLISECONDS, Schedulers.computation())
                        . take(Integer.MAX_VALUE). subscribe (e ::onNext);
            }
        }).observeOn (Schedulers.newThread());

        observable .subscribe(subscriber1) ;
        observable.subscribe(subscriber2) ;

        Thread.sleep(100L) ;
    }

    // do操作符 ，用于添加监听
    @Test
    public  void test04() {
        Observable.just(" Hello World")
                .doOnNext(onNext->{
                    System.out.println (" doOnNext : "+onNext) ;
                }).
                doAfterNext(afterNext->{
                    System.out.println (" afterNext : "+afterNext) ;
                })
                .doOnError(onError->{
                    System.out.println (" onError : "+onError) ;
                })
                .doOnComplete(new Action() {
                    @Override
                    public void run() throws Exception {
                        System.out.println (" OnComplete") ;
                    }
                })
                .subscribe(onNext->{
                    System.out.println ("订阅 onNext : ") ;
        });
    }

    @Test
    public  void test03() {
        Observable.just(" Hello World").subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable disposable) {

            }

            @Override
            public void onNext(@NonNull String s) {

            }

            @Override
            public void onError(@NonNull Throwable throwable) {

            }

            @Override
            public void onComplete() {

            }
        });
    }
    @Test
    public  void test02() {
        Observable.just(" Hello World").subscribe(
        onNext -> {
            System.out.println("onNext");
        }, onError -> {
                    System.out.println("onError");
        }, new Action() {
            @Override
            public void run() throws Exception {
                System.out.println("onComplete");
            }
        });
    }

    @Test
    public  void test01() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
                e.onNext (" Hello World");
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                System. out . println(s) ;
            }
        });
    }
}
